package org.qb.mapreduce.reduceJoinDemo;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

public class ReduceJoinReducer extends Reducer<Text,ReduceJoinBean,ReduceJoinBean, NullWritable> {

    @Override
    protected void reduce(Text key, Iterable<ReduceJoinBean> values, Reducer<Text, ReduceJoinBean, ReduceJoinBean, NullWritable>.Context context) throws IOException, InterruptedException {

        ArrayList<ReduceJoinBean> orderBeanList = new ArrayList<>();
        ReduceJoinBean pdBean = new ReduceJoinBean();

        for (ReduceJoinBean value : values) {

            if ("order".equals(value.getFlag())){

                ReduceJoinBean bean = new ReduceJoinBean();

                //copy value
                try {
                    BeanUtils.copyProperties(bean,value);
                } catch (IllegalAccessException | InvocationTargetException e) {
                    throw new RuntimeException(e);
                }

                // add record to the orderBeanList
                orderBeanList.add(bean);

            }else {
                //copy value
                try {
                    BeanUtils.copyProperties(pdBean,value);
                } catch (IllegalAccessException | InvocationTargetException e) {
                    throw new RuntimeException(e);
                }

            }

        }

        for (ReduceJoinBean beanList : orderBeanList) {
            beanList.setProductName(pdBean.getProductName());
            context.write(beanList,NullWritable.get());
        }
    }
}
